package com.atguigu.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {
        new DimApp().init(
                3001,
                2,
                "DimApp",
                Constant.TOPIC_ODS_DB
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        SingleOutputStreamOperator<TableProcess> tpProcess = readTableProcess(env);
        tpProcess = createDimTable(tpProcess);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connectedStream = connect(etledStream, tpProcess);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = delNoNeedCol(connectedStream);
        writeToPhoenix(resultStream);

    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNoNeedCol(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connectedStream) {
        return connectedStream
                .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                        JSONObject data = t.f0; // 维度数据
                        List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));// 配置信息
                        // 删除 data 中的那些字段:在SinkColumns中不存在的.
                        data.keySet().removeIf(key -> !columns.contains(key) && !"op_type".equals(key));

                        return t;
                    }
                });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> dataStream,
            SingleOutputStreamOperator<TableProcess> confStream) {
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = confStream.broadcast(tpStateDesc);
        return dataStream
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {

                    private HashMap<String, TableProcess> tpMap;

                    //TODO 解决数据先来，配置后来的问题
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 先去预加载所有配置. 存储到一个 Map 中 key: obj.getString("table") + ":ALL" value:TableProcess
                        // open方法中还不能直接访问状态
                        // 通过普通的 jdbc 来读所有配置信息
                        Connection conn = JdbcUtil.getMysqlConnection();
                        // select * from table_process  实现一个通用的查询工具类
                        String sql = "select * from edu_config.table_process";
                        List<TableProcess> list = JdbcUtil.queryList(conn, sql, null, TableProcess.class);
                        // 把 list 中的配置数据转存到 HashMap
                        tpMap = new HashMap<>();
                        for (TableProcess tp : list) {
                            if ("dim".equals(tp.getSinkType())) {
                                tpMap.put(tp.getSourceTable() + ":" + tp.getSourceType(),tp);
                            }
                        }
                    }

                    @Override
                    public void processElement(JSONObject obj, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        String key = obj.getString("table") + ":ALL";
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        TableProcess tp = state.get(key);
                        if (tp == null) {
                            tp = tpMap.get(key);
                        }
                        if (tp != null) {
                            JSONObject data = obj.getJSONObject("data");
                            data.put("op_type", obj.getString("type"));
                            out.collect(Tuple2.of(data, tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcess tp,
                                                        Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key = tp.getSourceTable() + ":" + tp.getSourceType();
                        state.put(key, tp);
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> createDimTable(SingleOutputStreamOperator<TableProcess> tpProcess) {
        return tpProcess
                //TODO 过滤dim数据
                .filter(tp -> "dim".equals(tp.getSinkType()))
                .process(new ProcessFunction<TableProcess, TableProcess>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        JdbcUtil.close(conn);
                    }

                    @Override
                    public void processElement(TableProcess tp, Context ctx, Collector<TableProcess> out) throws Exception {
                        //TODO 拼接 建表语句
                        String op = tp.getOp();
                        StringBuilder sql = null;

                        if ("r".equals(op) ||"c".equals(op)) {
                            sql = getCreateTableSql(tp);
                        } else if ("d".equals(op)) {
                            sql = getDelTableSql(tp);
                        } else {
                            // u : 先删
                            PreparedStatement ps = conn.prepareStatement(getDelTableSql(tp).toString());
                            ps.execute();
                            ps.close();
                            // 后建
                            sql = getCreateTableSql(tp);
                        }

                        //TODO 获取预处理语句
                        PreparedStatement ps = conn.prepareStatement(sql.toString());

                        //TODO 执行
                        ps.execute();

                        //TODO 关闭预处理
                        ps.close();

                        out.collect(tp);
                    }

                    private StringBuilder getDelTableSql(TableProcess tp) {
                        return new StringBuilder("drop table " + tp.getSinkTable());
                    }

                    private StringBuilder getCreateTableSql(TableProcess tp) {
                        StringBuilder sql = new StringBuilder();

                        sql
                                .append("create table if not exists ")
                                .append(tp.getSinkTable())
                                .append("(")
                                .append(tp.getSinkColumns().replaceAll("[^,]+", "$0 varchar"))
                                .append(", constraint pk primary key (")
                                .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                                .append("))")
                                .append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend());   // .. null
                        System.out.println("维度建表语句: " + sql);
                        return sql;
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
                .hostname( "hadoop162" )
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username ( "root")
                .password ( "aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        return env
                .fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        String op =  obj.getString("op");
                        TableProcess tp = null;
                        if ("d".equals(op)) {
                            tp = obj.getObject("before", TableProcess.class);
                        } else {
                            tp = obj.getObject("after", TableProcess.class);
                        }

                        tp.setOp(op);
                        return tp;
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        try {
                            JSONObject obj = JSON.parseObject(value.replaceAll("bootstrap-", ""));
                            String type = obj.getString("type");
                            String data = obj.getString("data");
                            // 证明 json 格式是对的
                            return "edu".equals(obj.getString("database"))
                                    && ("insert".equals(type) || "update".equals(type))
                                    && data != null
                                    && data.length() > 2;

                        } catch (Exception e) {
                            System.out.println("你的 json 格式数据异常: " + value);
                            return false;
                        }
                    }
                })
                .map(json -> JSON.parseObject(json.replaceAll("bootstrap-", "")));
    }
}
